package com.qf.qfschedule.service.impl;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.qf.qfleadnewsmodel.commons.consts.RedisConst;
import com.qf.qfleadnewsmodel.enums.TaskTypeEnum;
import com.qf.qfleadnewsmodel.schedule.dtos.Task;
import com.qf.qfleadnewsmodel.schedule.pojos.Taskinfo;
import com.qf.qfleadnewsmodel.schedule.pojos.TaskinfoLogs;
import com.qf.qfschedule.mapper.TaskInfoMapper;
import com.qf.qfschedule.service.TaskinfoLogsService;
import com.qf.qfschedule.service.TaskinfoService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.UnsupportedEncodingException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Stream;

@Service
@Slf4j
public class TaskinfoServiceImpl extends ServiceImpl<TaskInfoMapper, Taskinfo> implements TaskinfoService {

    @Autowired
    private ApplicationContext ioc;

    @Autowired
    private TaskinfoLogsService taskinfoLogsService;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Override
    public boolean add(Task task) {

        Taskinfo taskinfo = ioc.getBean(TaskinfoService.class).save2DB(task);

        if (taskinfo != null){
            //将数据库的信息保存到redis中
            save2Redis(taskinfo);

            //表示任务添加成功
            return true;
        }
        //表示任务添加失败
        return false;
    }

    /**
     * 将数据库的任务信息保存到redis中，提供后续消费使用
     * @param taskinfo
     */
    private void save2Redis(Taskinfo taskinfo) {
        //先获取到未来5分钟的时间点
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);
        //未来5分钟对应的时间毫秒值
        long future5MinLong = calendar.getTime().getTime();
        //任务执行时间毫秒值
        long taskLong = taskinfo.getExecuteTime().getTime();

        if (taskLong <= future5MinLong){
            //最近要执行的任务，需要加入redis
            //判断任务是否要立即执行？ 是--存入即时消费的队列 否-存入预热队列（后续通过定时任务，从预热队列中迁移到即时消费队列）

            //获取当前系统时间
            long nowLong = System.currentTimeMillis();

            if (taskLong <= nowLong){
                //存入即时消费队列
                String key = buildKey(taskinfo, RedisConst.LIST_TYPE);
                redisTemplate.opsForList().leftPush(key, JSON.toJSONString(taskinfo));
            }else{
                //存入预热队列
                String key = buildKey(taskinfo, RedisConst.ZSET_TYPE);
                redisTemplate.opsForZSet().add(key,JSON.toJSONString(taskinfo),taskLong);
            }
        }

    }

    /**
     *  根据任务类型、优先级和队列类型构建一个 key
     * @param taskinfo
     * @param type
     * @return
     */
    private String buildKey(Taskinfo taskinfo, String type) {
        Integer taskType = taskinfo.getTaskType(); // 1001
        Integer priority = taskinfo.getPriority(); //1
        String prefix = "";

        if (type == RedisConst.LIST_TYPE){
            prefix = RedisConst.NOW_CONSUME_TASK_PREFIX; //TASK:NOW:
        }else{
            prefix = RedisConst.FUTURE_CONSUME_TASK_PREFIX; // TASK:FUTURE:
        }

        return prefix + taskType + "_" + priority; // TASK:NOW:1001_1    TASK:FUTURE:1001_1
    }

    /**
     * 保存任务到数据库表中
     *  将数据保存到info表和logs表中，而且要保障这两个表的保存动作是一个事务
     * @param task
     */
    @Transactional
    public Taskinfo save2DB(Task task) {
        //保存到info
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        //对于executeTime额外处理
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));

        //保存
        save(taskinfo);

        //保存到logs
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        //将info中的数据拷贝到logs中
        BeanUtils.copyProperties(taskinfo, taskinfoLogs);
        //初始状态
        taskinfoLogs.setStatus(0);
        //初始版本号
        taskinfoLogs.setVersion(0);

        //保存
        taskinfoLogsService.save(taskinfoLogs);

        return taskinfo;
    }

    /**
     * 这个任务迁移是每隔1s做一次，频度高的，这边加读锁。在数据迁移方
     * @see #db2Redis() 加写锁
     */
    @Override
    @Transactional
    public Taskinfo consumeTask(int taskType, int priority) {
        Taskinfo taskinfo = new Taskinfo();
        taskinfo.setTaskType(taskType);
        taskinfo.setPriority(priority);
        String key = buildKey(taskinfo, RedisConst.LIST_TYPE);

        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisConst.DATA_MOVE_AND_CONSUME_READ_WRITE_LOCK);
        RLock lock = readWriteLock.readLock();

        try {
            lock.lock();
            //获取这个key对应List结构的最右边一条记录（我们是从左边存入，所以是从右边取出）
            String taskJson = redisTemplate.opsForList().rightPop(key);

            if (StringUtils.isNotBlank(taskJson)) {
                taskinfo = JSON.parseObject(taskJson, Taskinfo.class);

                //重新维护数据库中这个任务数据(① 删除taskinfo表中对应记录 ② 修改log表中状态值)
                //① 删除taskinfo表中对应记录
                removeById(taskinfo.getTaskId());

                //② 修改log表中状态值
                TaskinfoLogs logs = taskinfoLogsService.getById(taskinfo.getTaskId());
                logs.setStatus(1); //1表示已被执行

                //使用mybatis-plus的乐观锁插件自动维护版本号（注入导入插件）
                //logs.setVersion(logs.getVersion()+1);

                taskinfoLogsService.updateById(logs);

                return taskinfo;
            }
        }finally {
            lock.unlock();
        }

        return null;
    }

    /**
     * 将预热队列zset中的任务数据，迁移到list即时消费队列中
     *  定时任务，在集群环境下，只允许有一个程序在执行，那么我们必须只能加 - 分布式互斥锁
     *
     */
    @Scheduled(cron = "0/5 * * * * ? ")
    public void zset2List() throws UnsupportedEncodingException {
        RLock lock = redissonClient.getLock(RedisConst.ZSET_2_LIST_LOCK);

        try {
            log.info("----------zset数据迁移到list开始,时间为：{}", LocalDateTime.now());
            lock.lock();
         /*
            scan方法返回回来的符合统配规则的键肯能存在重复值的，所以这里我们使用Set集合实现去重
         */
            Set<String> futureKeys = null;

            try (RedisConnection conn = redisTemplate.getConnectionFactory().getConnection();) {
                //到redis中查找所有 TASK:FUTURE: 前缀相关的键
                ScanOptions options = ScanOptions.scanOptions()
                        //要查找的键的统配字符串
                        .match(RedisConst.FUTURE_CONSUME_TASK_PREFIX + "*")
                        //要查找的数量
                        .count(1000)
                        .build();
                //查询到的所有符合规则的键都封装在这个游标对象中
                Cursor<byte[]> cursor = conn.scan(options);

                if (cursor != null) {
                    futureKeys = new HashSet<>();
                    while (cursor.hasNext()) {
                        byte[] next = cursor.next();
                        String key = new String(next, "utf-8");
                        futureKeys.add(key);
                    }
                }
            }

            //找到了所有需要迁移的zset中的key
            if (futureKeys != null) {
                futureKeys.stream().forEach(futureKey -> {
                    //通过futureKey获取这个zset中对那个需要立即执行的任务数据
                    //获取符合分值范围的数据[0,当前系统时间]  -- 需要立即执行的任务
                    Set<String> values = redisTemplate.opsForZSet().rangeByScore(futureKey, 0, System.currentTimeMillis());

                    if (values != null && values.size() > 0) {
                        //将这些值从zset中移除
                        redisTemplate.opsForZSet().remove(futureKey, values.toArray());
                        //将这些值添加到list中

                        String nowKey = futureKey.replaceAll(RedisConst.FUTURE_CONSUME_TASK_PREFIX, RedisConst.NOW_CONSUME_TASK_PREFIX);
                        redisTemplate.opsForList().leftPushAll(nowKey, values.toArray(new String[]{}));
                    }
                });
            }
        }finally {
            lock.unlock();
        }
    }

    /**
     * 将数据库中的任务数据，迁移到redis中 (5分钟)
     */
    @Scheduled(cron = "0 0/5 * * * ? ")
    public void db2Redis(){
        log.info("----------数据库数据迁移到redis开始,时间为：{}", LocalDateTime.now());

        //迁移的目标数据是哪些？？？ 在未来5分钟之内要执行的任务
        //先获取到未来5分钟的时间点
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);
        Date time = calendar.getTime();

        Wrapper<Taskinfo> qr = Wrappers.lambdaQuery(Taskinfo.class)
                .lt(Taskinfo::getExecuteTime,time);

        /**
         * 这个任务迁移是每隔5分钟做一次，频度没有那么高的，这边加写锁。在任务消费方
         * @see #consumeTask(int taskType, int priority) 加读锁
         */
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisConst.DATA_MOVE_AND_CONSUME_READ_WRITE_LOCK);
        RLock lock = readWriteLock.writeLock();

        try {
            lock.lock();
            List<Taskinfo> taskinfos = list(qr);
            //先将之前redis的数据移除
            TaskTypeEnum[] taskTypes = TaskTypeEnum.values();
            Stream.of(taskTypes).forEach(taskType -> {
                int type = taskType.getTaskType();
                int priority = taskType.getPriority();
                Taskinfo info = new Taskinfo();
                info.setTaskType(type);
                info.setPriority(priority);

                String nowKey = buildKey(info, RedisConst.LIST_TYPE);
                String futureKey = buildKey(info, RedisConst.ZSET_TYPE);

                //删除数据
                redisTemplate.delete(nowKey);
                redisTemplate.delete(futureKey);
            });

            //保存到redis
            taskinfos.forEach(info -> {
                save2Redis(info);
            });
        }finally {
            lock.unlock();
        }
    }
}
